package com.atguigu.gmall.realtime.app.dws;

import com.atguigu.gmall.realtime.app.func.KeywordUDTF;
import com.atguigu.gmall.realtime.beans.KeywordStats;
import com.atguigu.gmall.realtime.common.GmallConstant;
import com.atguigu.gmall.realtime.utils.ClickHouseUtil;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Author: Felix
 * Date: 2022/1/8
 * Desc: 关键词统计
 */
public class KeywordStatsApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //1.3 设置表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        /*
        //TODO 2.检查点设置
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(6000L);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000L);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));
        env.setStateBackend(new FsStateBackend(""));
        System.setProperty("HADOOP_USER_NAME","atguigu");
        */

        //注册自定义函数
        tableEnv.createTemporarySystemFunction("ik_analyze", KeywordUDTF.class);

        //TODO 3.从Kafka中读取数据创建动态表
        String topic = "dwd_page_log";
        String groupId = "keyword_stats_group";
        tableEnv.executeSql("create table page_log(" +
            "common MAP<STRING, STRING>," +
            "page MAP<STRING, STRING>," +
            "ts BIGINT," +
            "rowtime AS TO_TIMESTAMP(FROM_UNIXTIME(ts/1000, 'yyyy-MM-dd HH:mm:ss'))," +
            "WATERMARK FOR rowtime AS rowtime - INTERVAL '3' SECOND" +
            ")WITH (" + MyKafkaUtil.getKafkaDDL(topic, groupId) + ")");

        //TODO 4.从表中过滤搜索行为
        Table searchTable = tableEnv.sqlQuery("select page['item'] fullword,rowtime from page_log " +
            " where page['page_id']='good_list' and page['item'] is not null");

        //TODO 5.使用自定义的UDTF函数对搜索的内容进行分词
        Table splitTable = tableEnv.sqlQuery("select keyword,rowtime from "+searchTable+",LATERAL TABLE(ik_analyze(fullword)) as t(keyword)");

        //TODO 6.分组、开窗、聚合计算
        Table keywordStatsSearch  = tableEnv.sqlQuery("select keyword,count(*) ct, '"
            + GmallConstant.KEYWORD_SEARCH + "' source ," +
            "DATE_FORMAT(TUMBLE_START(rowtime, INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') stt," +
            "DATE_FORMAT(TUMBLE_END(rowtime, INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') edt," +
            "UNIX_TIMESTAMP()*1000 ts from   "+splitTable
            + " GROUP BY TUMBLE(rowtime, INTERVAL '10' SECOND ),keyword");

        //TODO 7.将动态表转换为流
        DataStream<KeywordStats> keywordStatsDS = tableEnv.toAppendStream(keywordStatsSearch, KeywordStats.class);

        //TODO 8.将流中的数据写到ClickHouse中
        keywordStatsDS.print(">>>>>");
        keywordStatsDS.addSink(
            ClickHouseUtil.getJdbcSink("insert into keyword_stats_0701(keyword,ct,source,stt,edt,ts) values(?,?,?,?,?,?)")
        );

        env.execute();
    }
}
